package com.lry.proxy;

import com.lry.common.CommonConstants;
import com.lry.exception.BizException;
import com.lry.filter.FilterFactory;
import com.lry.protocol.*;
import com.lry.rpc.config.NettyClientConfig;
import com.lry.rpc.config.RpcReferenceBean;
import com.lry.rpc.netty.MsgType;
import com.lry.rpc.netty.RpcContext;
import com.lry.rpc.netty.client.ChannelWrapper;
import com.lry.rpc.netty.client.ClientCache;
import com.lry.rpc.netty.client.MessageSender;
import com.lry.util.CollectionUtils;
import com.lry.util.StringUtils;
import io.netty.channel.Channel;
import lombok.extern.slf4j.Slf4j;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Future;
import java.util.concurrent.TimeoutException;

import static com.lry.common.CommonConstants.COLON;
import static com.lry.rpc.netty.client.ClientCache.ID_GEN;

@Slf4j
public class RpcInvoker implements Invoker, MessageSender {

    private ChannelWrapper channelWrapper;
    private RpcCallback rpcCallback;

    public void setRpcCallback(RpcCallback rpcCallback) {
        this.rpcCallback = rpcCallback;
    }

    public RpcInvoker(ChannelWrapper channelWrapper){
        this.channelWrapper = channelWrapper;
    }

    @Override
    public Object invoke(Invocation invocation) throws TimeoutException {
        if(null==rpcCallback){
            return sendSyncRequest(channelWrapper.getChannel(),invocation);
        }else{
            return sendAsyncRequest(channelWrapper.getChannel(),invocation,rpcCallback);
        }
    }

    @Override
    public boolean ipPortEquals(String ip, Integer port) {
        if(ip.equals(channelWrapper.getIp())&&port.equals(channelWrapper.getPort())){
            return true;
        }
        return false;
    }

    @Override
    public int getWeight() {
        return channelWrapper.getWeight();
    }

    @Override
    public void updateWeight(Integer weight) {
        channelWrapper.setWeight(weight);
    }

    @Override
    public String getIps() {
        return channelWrapper.getIp()+COLON+channelWrapper.getPort();
    }

    @Override
    public boolean isAvailable() {
        return channelWrapper.getChannel().isActive();
    }


    @Override
    public Object sendSyncRequest(Channel channel, Object message) throws TimeoutException   {
        return sendSyncRequest(channel,message,0);
    }

    @Override
    public Object sendSyncRequest(Channel channel,Object message,int tryCount)  throws TimeoutException {
        RpcInvocation rpcInvocation = (RpcInvocation) message;
        RpcReferenceBean referenceBean = ClientCache.getReferenceBean(rpcInvocation);
        if(tryCount>=referenceBean.getRetries()){
            throw new TimeoutException(String.format("tryCount=%s,all timeout",tryCount));
        }

        long timeout = referenceBean.getTimeout();

        int requestId = ID_GEN.getAndIncrement();
        RpcMessage rpcMessage = new RpcMessage(requestId, MsgType.RPC_REQUEST.getCode(), message);
        rpcMessage.setSerializerStr(referenceBean.getSerializer());
        rpcMessage.setCompressorStr(referenceBean.getCompressor());

        FilterFactory.executeAllClientFilter(rpcMessage);

        channel.writeAndFlush(rpcMessage);

        RpcMessage response = new RpcMessage(requestId);
        ClientCache.putToPending(requestId,response);

        synchronized (response){
            try {
                response.wait(timeout);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        //只要接受到了服务端的数据，这个body就不是null
        //如果指定timeout没有接到服务端的消息，或者服务端执行出现了异常，都要进行重试指定retries次
        if(null==response.getBody()){
            return sendSyncRequest(channel,message,tryCount+1);
        }else{
            RpcResult rpcResult = (RpcResult) response.getBody();
            if(rpcResult.getCode()== ResultEnums.FAIL.getCode()){
                return sendSyncRequest(channel,message,tryCount+1);
            }else if(rpcResult.getCode()== ResultEnums.RETRY_FAIL.getCode()){
                //接到下游的重试错误，则不再重试了
                throw new TimeoutException(String.format("接到下游的重试错误，则不再重试了，tryCount=%s,all timeout",tryCount));
            }else{
                return rpcResult.getObject();
            }
        }
    }

    @Override
    public Future<Object> sendAsyncRequest(Channel channel, Object message, RpcCallback<Object> callback)  {

        Future<Object> future = CompletableFuture.supplyAsync(()->{
            try {
                Object obj = sendSyncRequest(channel,message);
                callback.success(obj);
                return obj;
            } catch (Exception e) {
                callback.fail(e);
                e.printStackTrace();
            }
            return null;
        });
        return future;
    }


}
